package com.fangsheng.zookeeper.lock;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * @创建人: 放生
 * @创建时间: 2021/11/3
 * @描述:
 */
public class WatchCallBack implements Watcher, AsyncCallback.StringCallback ,AsyncCallback.Children2Callback ,AsyncCallback.StatCallback{
    ZooKeeper zk ;
    String threadName;
    CountDownLatch cc = new CountDownLatch(1);
    String pathName;

    public String getPathName() {
        return pathName;
    }

    public void setPathName(String pathName) {
        this.pathName = pathName;
    }

    public String getThreadName() {
        return threadName;
    }

    public void setThreadName(String threadName) {
        this.threadName = threadName;
    }

    public ZooKeeper getZk() {
        return zk;
    }

    public void setZk(ZooKeeper zk) {
        this.zk = zk;
    }

    /**
     * 该方法会并发创建N个节点/lock，因为之前在ZKUtils中设置了"192.168.10.11:2181,192.168.10.21:2181,192.168.10.31:2181/testLock"
     * 所以是创建在/testLock下面的会是/testLock/lock的形式，在方法zk.create("/lock",threadName.getBytes(),
     *   ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,this,"abc");中
     *      第一个参数是节点/lock，
     *      第二个参数是：这个节点存的数据，这里是存的线程的名字，
     *      第三个参数是：ZooDefs.Ids.OPEN_ACL_UNSAFE表示的是权限开放，
     *      第四个参数表示创建的是一个队临时列节点，
     *      第五个参数是 StringCallback回调，会回调下面processResult(int rc, String path, Object ctx, String name)方法
     *      第六个参数是 给回调的传参 这里不用所以任意传了一个 abc
     *   因为是一个临时队列节点所以最终创建的节点是带序列号的 /testLock/lock0000000001,/testLock/lock0000000002 ...../testLock/lockxxxxxxxxx
     *
     *   则10个并发过来会创建10个节点分别是/testLock/lock0000000000，/testLock/lock0000000001，/testLock/lock0000000009
     *   这个序列号是自增的，从 0 开始
     *   10个创建时都会回调StringCallback回调processResult(int rc, String path, Object ctx, String name)方法
     *   具体的描述见这个方法
     *
     */
    public void tryLock(){
        try {

            System.out.println(threadName + "  create....");
//            if(zk.getData("/"))
            // 创建一个临时的 队列节点
            // 注意创建 zk.create（）这个方法是异步的，所以会在cc.await() 阻塞住，但是第创建成功后会调用zk.create("/lock",threadName.getBytes(),
            // ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,this,"abc"); 方法中的回调，也就是processResult(int rc, String path,
            // Object ctx, String name) 方法
            zk.create("/lock",threadName.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL,this,"abc");

            cc.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     *  业务处理完删除 这个节点unLock()，然后就会执行process(WatchedEvent event)
     *  中的case NodeDeleted:  zk.getChildren("/",false,this ,"sdf");方法，
     *  这个方法又会回调 Children2Callback，从而有进入 processResult(int rc, String path,
     *  Object ctx, List<String> children, Stat stat) 这个方法，让下一个节点获取锁。。。
     */
    public void unLock(){
        try {
            zk.delete(pathName,-1);
            System.out.println(threadName + " over work....");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (KeeperException e) {
            e.printStackTrace();
        }
    }


    @Override
    public void process(WatchedEvent event) {


        //如果第一个哥们，那个锁释放了，其实只有第二个收到了回调事件！！
        //如果，不是第一个哥们，某一个，挂了，也能造成他后边的收到这个通知，从而让他后边那个跟去watch挂掉这个哥们前边的。。。
        switch (event.getType()) {
            case None:
                break;
            case NodeCreated:
                break;
            case NodeDeleted:
                zk.getChildren("/",false,this ,"sdf");
                break;
            case NodeDataChanged:
                break;
            case NodeChildrenChanged:
                break;
        }

    }

    /**
     *
     * @param rc 是一个下标 0
     * @param path 就是创建 /lock目录不带序列号的
     * @param ctx  上一个回调闯过来的值 这里是abc
     * @param name  是每次带有序列号的值如：/lock0000000001
     * 这里如果name有值则带有序列号的节点创建成功，同样这里是10并发过来都会执行，
     * 并且在这个方法中获取下面所有的节点 zk.getChildren("/",false,this ,"sdf");方法中的 “/”是指/testLock/lock
     *   会触发下面 Children2Callback 的回调 processResult(int rc, String path,
     *              Object ctx, List<String> children, Stat stat)
     *      具体的过程见他下面的方法
     */
    @Override
    public void processResult(int rc, String path, Object ctx, String name) {
        if(name != null ){
            System.out.println("rc==>"+rc);
            System.out.println("path==>"+path);
            System.out.println("ctx==>"+ctx);
            System.out.println("name==>"+name);
            System.out.println(threadName  +"  create node : " +  name );
            pathName =  name ;
            // 所有的线程获取他节点下的children，这个时候不需要监听所以watch FALSE
            // 但是回调是要写的，会回调下面的processResult(int rc, String path, Object ctx, List<String> children, Stat stat) 方法
            zk.getChildren("/",false,this ,"sdf");
        }

    }

    //getChildren  call back

    /**
     *
     * @param rc 是一个下标 0
     * @param path 就是创建 /lock目录不带序列号的
     * @param ctx  上一个回调传下来的值 sdf
     * @param children 所有上一个回调时候“/” 下面的所有的节点的集合
     * @param stat
     * 同样该方法也会10个线程都并发进来，并且每次给节点排序会是如下的顺序
     *         /lock0000000000，/lock0000000001，/lock0000000002，.... /lockxxxxxxxxxx
     *      int i = children.indexOf(pathName.substring(1));是去除  / 获取到当前lock0000000000（第一次进来）
     *      的下标，赋值给i,
     *      如果i==0，表示当前方法被执行的就是位于队列的第一个，也是序列号当前最小的一个
     *      第一个就获取锁，cc.countDown();这个时候一个线程的的tryLock()方法结束了，就会执行
     *      TestLock类中的业务逻辑代码，
     *      如果i ！= 0 则表示当前的线程操作的不是队列中最前的，则会
     *      阻塞住zk.exists("/"+children.get(i-1),this,this,"sdf")，该方法就获取他的上一个序列节点children.get(i-1)判断
     *      是否存在，并且watch它上一个序列，每一个都watch它的上一个，就相当于10个线程进来
     *      lock0000000000当前最先获取锁，而其他的lock0000000001 就会个他前面一个序列lock000000000添加watch，
     *      lock0000000002，就给他的前面一个序列lock0000000001添加watch.......，这样形成一个链表的watch，只有
     *      他前面的临时节点删除就会触发下一个节点获取锁。
     *
     */
    @Override
    public void processResult(int rc, String path, Object ctx, List<String> children, Stat stat) {

        //一定能看到自己前边的。。

//        System.out.println(threadName+"look locks.....");
//        for (String child : children) {
//            System.out.println(child);
//        }

        Collections.sort(children);
        int i = children.indexOf(pathName.substring(1));


        //是不是第一个
        if(i == 0){
            //yes
            System.out.println(threadName +" i am first....");
            try {
                zk.setData("/",threadName.getBytes(),-1);
                cc.countDown();

            } catch (KeeperException e) {
                e.printStackTrace();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }else{
            //no
            zk.exists("/"+children.get(i-1),this,this,"sdf");
        }

    }

    @Override
    public void processResult(int rc, String path, Object ctx, Stat stat) {
        //偷懒
    }


}
